package org.aceor.mddal.config.datasource.factory;

import com.google.common.base.Strings;
import com.thoughtworks.xstream.XStream;
import org.aceor.mddal.common.utils.RuntimeConfigHolder;
import org.aceor.mddal.common.zk.DynConfigClient;
import org.aceor.mddal.common.zk.DynConfigClientFactory;
import org.aceor.mddal.common.zk.listener.ConfigChangeListener;
import org.aceor.mddal.common.zk.object.ConfigDTO;
import org.aceor.mddal.config.datasource.Atom;
import org.aceor.mddal.config.datasource.AtomNode;
import org.aceor.mddal.config.datasource.DsConfig;
import org.aceor.mddal.config.datasource.GroupNode;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStream;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * Created by lxue on 16/3/27.
 */
public class DsConfigFactory {

    private static final Logger log = LoggerFactory.getLogger(DsConfigFactory.class);

    private static final String DS_CONFIG_BIZ_NAME = "ds";
    private static final String DS_CONFIG_DATA_ID = "dsConfig";
    //单例容器
    private volatile static RuntimeConfigHolder<DsConfig> dsConfigHolder = null;

    private static DynConfigClient configClient;

    public static RuntimeConfigHolder<DsConfig> buildDsConfig(String dsConfigPath, DsConfigChangeListener dsConfigChangeListener) throws Exception {
        if (null == dsConfigHolder) {
            synchronized (DsConfigFactory.class) {
                if (null == dsConfigHolder) {
                    dsConfigHolder = new RuntimeConfigHolder<DsConfig>();
                    dsConfigHolder.set(buildDsConfigInternal(dsConfigPath, dsConfigChangeListener));
                }
            }
        }
        return dsConfigHolder;
    }

    private static DsConfig buildDsConfigInternal(String dsConfigPath, DsConfigChangeListener dsConfigChangeListener) throws Exception {
        if (null == dsConfigChangeListener) {
            return parseDsConfigFromXml(DsConfigFactory.class.getResourceAsStream(dsConfigPath));
        } else {
            configClient = DynConfigClientFactory.getConfigClient();
            String dsConfigStr = null;
            try {
                dsConfigStr = fetchDsConfigStrFromZK();
                registerDsConfigChangeListener(dsConfigChangeListener);
                return parseDsConfigFromXml(dsConfigStr);
            } catch (Exception e) {
                //zk获取失败先尝试本地获取
                if (!Strings.isNullOrEmpty(dsConfigPath)) {
                    log.info("从zk获取dsConfig失败,尝试从本地获取,localDsConfigPath :{}", dsConfigPath);
                    return parseDsConfigFromXml(DsConfigFactory.class.getResourceAsStream(dsConfigPath));
                } else {
                    throw e;
                }
            }
        }
    }

    private static DsConfig parseDsConfigFromXml(Object source) {
        XStream xStream = new XStream();
        xStream.processAnnotations(new Class[]{DsConfig.class, GroupNode.class, AtomNode.class, Atom.class});
        DsConfig dsConfig;
        if (source instanceof String) {
            dsConfig = (DsConfig) xStream.fromXML((String) source);
        } else if (source instanceof InputStream) {
            dsConfig = (DsConfig) xStream.fromXML((InputStream) source);
        } else {
            throw new UnsupportedOperationException("暂时只支持String和InputStream类型");
        }
        dsConfig.init();
        return dsConfig;
    }

    private static void registerDsConfigChangeListener(DsConfigChangeListener dsConfigChangeListener) {
        configClient.registerListener(DS_CONFIG_BIZ_NAME, DS_CONFIG_DATA_ID, new ConfigChangeListener() {
            @Override
            public Executor getExecutor() {
                return Executors.newSingleThreadExecutor();
            }

            @Override
            public void receiveConfig(ConfigDTO configDTO) {
                log.info("获取到最新的DsConfig配置, dsConfig :{}", configDTO.getConfig());
                dsConfigHolder.set(parseDsConfigFromXml(configDTO.getConfig()));
                dsConfigChangeListener.onChange(dsConfigHolder.get());
            }
        });
    }

    private static String fetchDsConfigStrFromZK() throws Exception {
        try {
            return configClient.getConfig(DS_CONFIG_BIZ_NAME, DS_CONFIG_DATA_ID);
        } catch (Exception e) {
            log.error("ZK获取数据失败!bizName :{}, dataId :{}, e :{}", DS_CONFIG_BIZ_NAME, DS_CONFIG_DATA_ID, ExceptionUtils.getStackTrace(e));
            throw e;
        }
    }

    public static void reset() {
        if (null != dsConfigHolder) {
            dsConfigHolder.set(null);
            dsConfigHolder = null;
        }
    }

    public interface DsConfigChangeListener {
        void onChange(DsConfig dsConfig);
    }

}
